package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.*;
import org.junit.Test;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerQuickStart {

    @Test
    public void sendMsg() throws ExecutionException, InterruptedException {
        // 2. 完成kafka配置
        Properties properties = new Properties();
            // 2.1 完成kafka的ip和端口
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"162.14.113.165:9092");
            // 2.2 配置key的序列话器
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
            // 2.3 配置value的序列化器
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        // 1. 创建生产的核心对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        for (int i = 0 ; i< 100 ; i++) {
            // 3. 发送消息
            ProducerRecord<String,String> record = new ProducerRecord<>("order","hello","world" + i );
            RecordMetadata recordMetadata = producer.send(record).get();
            System.out.println(recordMetadata);
        }

        // 4. 关闭连接
        producer.close();
    }

    @Test
    public void sendMsgAsync() throws ExecutionException, InterruptedException {
        // 2. 完成kafka配置
        Properties properties = new Properties();
        // 2.1 完成kafka的ip和端口
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"162.14.113.165:9092");
        // 2.2 配置key的序列话器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        // 2.3 配置value的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        // 默认为1 表示确认机制，0-不确认，数据最快，不过有消息丢失风险，1，主跟随着返回确认 all,主跟随着和随从跟随者都返回确认机制
        properties.put(ProducerConfig.ACKS_CONFIG,"all");

        // 如果发送失败则重新发送，极致是十次（每次重发需等待100ms）
        properties.put(ProducerConfig.RETRIES_CONFIG,10);

        // 1. 创建生产的核心对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        for (int i = 0 ; i< 100 ; i++) {
            // 3. 发送消息
            ProducerRecord<String,String> record = new ProducerRecord<>("order","hello","world" + i );
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    if(null != e){
                        e.printStackTrace();
                    }

                    System.out.println(metadata.offset());
                }
            });

        }

        // 4. 关闭连接
        producer.close();
    }
}
